package com.example.spider.spider;

import cn.hutool.core.util.StrUtil;
import com.example.spider.config.SpiderMetrics;
import com.example.spider.event.DropProxyEvent;
import com.example.spider.event.DropTaskEvent;
import com.example.spider.event.RetriesLimitEvent;
import com.example.spider.proxy.ProxyProvider;
import com.example.spider.proxy.SpiderContextHolder;
import com.example.spider.proxy.core.ProxyInfo;
import com.example.spider.task.AbstractSpiderTask;
import com.example.spider.task.SpiderTaskQueue;
import com.example.spider.task.Task;
import com.example.spider.task.limit.LimitStrategy;
import com.example.spider.util.FileBasedMap;
import jakarta.annotation.Nullable;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.shoulder.core.util.ContextUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.client.*;

import java.io.IOException;
import java.net.*;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 使用restTemplate 的爬虫
 * 1. 决定代理
 * 2. 决定是否限流，是则【更换代理】？直接等待？放弃任务？
 * 获取xxx业务最久没有使用过的代理【要保存代理对每个业务成功执行最后一次记录】
 * 业务维度的代理池隔离
 *
 * @author lym
 */
@Slf4j
@Component
public class RestSpider implements Spider {

    @Autowired
    private RestTemplate rest;

    @Autowired
    private ProxyProvider proxyProvider;

    @Autowired
    @Qualifier("spiderMetrics")
    private SpiderMetrics metrics;

    private final ConcurrentHashMap<String, Long> doSpiderHistory = new ConcurrentHashMap<>(16);

    private static <T> T handleServerError(AbstractSpiderTask<T> task, ResponseEntity<T> responseEntity, HttpServerErrorException e) {
        Set<String> limitRateUse503 = Set.of(
                "kuaidaili", "cdn-bo2.mangguo-youku.com:5278", "asp-bo2.mangguozy-videos.com:5278"
        );
        boolean return503 = e.getStatusCode() == HttpStatus.SERVICE_UNAVAILABLE
                || (e instanceof HttpServerErrorException.ServiceUnavailable && StringUtils.contains(e.getMessage(), "503 Service Temporarily Unavailable"));
        if (return503 && limitRateUse503.contains(task.getDomain())) {
            //503 建议不重试，部分任务限流也返回该值 有的网站限流也返回 503，等待个性化异常处理
            task.getLimitStrategy().doLimit(task);
            log.warn("RETRY 503 (count={}) cause - Ex({},cause={}) task:{} ExMsg={}.", task.getRetryNum(), e.getClass().getSimpleName(),
                    Optional.ofNullable(e.getCause()).map(ex -> ex.getClass().getSimpleName()).orElse("null"),
                    task, e.getMessage());
            return null;
        }
        return dropTask(task, responseEntity, e);
    }

    /**
     * 一定无法处理的异常，只能丢弃该任务
     */
    @Nullable
    public static <T> T dropTask(Task<T> task, ResponseEntity<T> responseEntity, Exception e) {
        String exName = e.getClass().getSimpleName();
        String msg = Optional.of(e).map(Throwable::getCause).map(Throwable::getMessage).orElse(e.getMessage());
        String causeExName = Optional.of(e).map(Throwable::getCause).map(Object::getClass).map(Class::getSimpleName).orElse(null);

        // 这两个地址会返回 503
        log.error("NO RETRY (cost={}ms) proxy={} Ex({},cause={}) msg=[{}] task:{}.", ((AbstractSpiderTask<T>) task).getRequestCost(),
                !"127.0.0.1".equals(task.getActuallyProxy().getIp()), exName, causeExName, msg,
                task.getUrl(), e);
        ContextUtils.publishEvent(new DropTaskEvent(task, e));

        String exKey = task.getClass().getSimpleName() + "___" + exName + "___" + causeExName + "___" + msg;
        FileBasedMap<ConcurrentHashMap<String, AtomicInteger>, String, AtomicInteger> spider_ex_info = ContextUtils.getBean("spider_ex_info");
        spider_ex_info.computeIfAbsent(exKey, k -> new AtomicInteger()).incrementAndGet();

        // null 不会处理，该任务丢失
        return null;
    }

    /**
     * 可以重试的异常
     */
    @Nullable
    public static <T> T retry(Exception e, Task<T> task) {
        // 重试不打堆栈
        log.warn("RETRY({} proxy={} count={}) cause - Ex({},cause={}) url:{} ExMsg={}.", task.getClass().getSimpleName(),
                !"127.0.0.1".equals(task.getActuallyProxy().getIp()), task.getRetryNum(), e.getClass().getSimpleName(),
                Optional.ofNullable(e.getCause()).map(ex -> ex.getClass().getSimpleName()).orElse("null"),
                task.getUrl(), e.getMessage());
        task.setRetryNum(task.getRetryNum() + 1);
        if (task.canRetry()) {
            SpiderTaskQueue.putTask((AbstractSpiderTask<?>) task);
        } else {
            ContextUtils.publishEvent(new RetriesLimitEvent(task, e));
        }
        return null;
    }

    public <T> T onException(Task<T> task, ResponseEntity<T> responseEntity, Exception e) {
        Set<Class<? extends Exception>> canNotRetryEx = Set.of(
                UnknownHostException.class, HttpRetryException.class, NoRouteToHostException.class, URISyntaxException.class, ProtocolException.class
        );
        if (e instanceof ResourceAccessException) {
            IOException ioException = (IOException) e.getCause();
            assert ioException != null;
            // UnknownHostException HttpRetryException NoRouteToHostException  URISyntaxException  ProtocolException
            //,cause=SSLException) msg=[Unsupported or unrecognized SSL message] canRetry？
            if (canNotRetryEx.contains(ioException.getClass())) {
                // 加黑名单不要重试了
                return task.onNoRetryEx(e);
            }
            if (ioException instanceof ConnectException && "Connection refused: no further information".equals(ioException.getMessage())) {
                // 无法访问，绝大部分情况不可重试，可以往往下载速度也很低
                return retry(e, task);
            }
            if (ioException instanceof SocketException && "Malformed reply from SOCKS server".equals(ioException.getMessage())) {
                // 代理类型错了，如代理服务器实际为HTTP代理，使用了SCOKS方式请求，建议纠正代理类型，这里用其他代理重试
                ContextUtils.publishEvent(new DropProxyEvent(task, e));
                return retry(e, task);
            }
            if (ioException instanceof SocketTimeoutException || ioException instanceof SocketException) {
                // 可重试 "Connection timed out", "Read timed out", "Connection refused", "Connection reset", "Connect timed out"
                if (StrUtil.containsAny(ioException.getMessage(), "timed out", "Connection refused", "Connection reset")) {
                    return retry(e, task);
                }
            }
            //if (ioException.getMessage().contains("I/O error on ") && ioException.getCause() instanceof UnknownHostException) {
            // 全都是 UnknownHostException ，检查本机是否有网，网线正常？换DNS、一会再试
            //}
            // Connection reset
            if (ioException.getMessage().contains("Unable to tunnel")) {
                ContextUtils.publishEvent(new DropProxyEvent(task, e));
                return retry(e, task);
            }
            // Unable to tunnel through proxy. Proxy returns "HTTP/1.1 503 Service Unavailable

            //SSLHandshakeException: Remote host terminated the handshake 【DEBUG 断点容易使得握手超时
            return dropTask(task, responseEntity, e);
        } else if (e instanceof HttpServerErrorException) {
            // 5xx
            return handleServerError((AbstractSpiderTask<T>) task, responseEntity, (HttpServerErrorException) e);
        } else if (e instanceof HttpClientErrorException.Forbidden || e instanceof HttpClientErrorException.NotFound || e instanceof HttpClientErrorException.TooManyRequests) {
            task.onNoRetryEx(e);
            return dropTask(task, responseEntity, e);
        } else if (e instanceof HttpClientErrorException.Unauthorized) {
            // 丢弃代理
            ContextUtils.publishEvent(new DropProxyEvent(task, e));
            return retry(e, task);
        } else if (e instanceof RestClientException) {
            if (e.getCause() instanceof SocketTimeoutException || e.getCause() instanceof SocketException) {
                return retry(e, task);
            }
            return dropTask(task, responseEntity, e);
        }
        throw new RuntimeException(e);
    }

    @Override
    public <T> T doSpider(AbstractSpiderTask<T> task) {
        // 上下文
        SpiderContextHolder.setTask(task);
        boolean success = false;
        T result = null;
        long startTime = System.currentTimeMillis();
        long endTime = startTime;
        Throwable t = null;
        ResponseEntity<T> responseEntity = null;
        metrics.activeTask(task);
        int retryCount = task.getRetryNum();
        try {
            String url = task.getUrl();
            log.debug("starting spider To {}", url);

            // 决策代理
            ProxyInfo proxyInfo = proxyInfo = proxyProvider.getProxy();
            task.setActuallyProxy(proxyInfo);
            assert task.getActuallyProxy() != null;

            LimitStrategy limitStrategy = task.getLimitStrategy();
            if (limitStrategy != null && limitStrategy.needLimit(task)) {
                limitStrategy.doLimit(task);
                // 限流
                return null;
            }

            // requestHeader
            RequestCallback requestCallback = request ->
                    task.getHeaders().forEach((k, v) -> request.getHeaders().add(k, v));
            // requestMethod
            HttpMethod requestMethod = Optional.ofNullable(task.getRequestMethod())
                    .filter(StringUtils::isNotBlank)
                    .map(HttpMethod::valueOf)
                    .orElse(HttpMethod.GET);

            // respType
            ResponseExtractor<ResponseEntity<T>> responseExtractor = rest.responseEntityExtractor(task.getResultType());

            Object[] params = task.getParams();
            startTime = System.currentTimeMillis();
            if (params != null && params.length > 0) {
                responseEntity = rest.execute(url, requestMethod, requestCallback, responseExtractor, params);
            } else {
                responseEntity = rest.execute(url, requestMethod, requestCallback, responseExtractor);
            }
            success = true;
            endTime = System.currentTimeMillis();
            long requestCost = endTime - startTime;
            proxyInfo.success(requestCost);
            task.setRequestCost(requestCost);

            result = responseEntity.getBody();
            onResponse(task, result, null, endTime);

            // TODO 返回 headers
            // responseEntity.getHeaders();
            // TODO 返回 ex 业务自己处理
            log.debug("finished spider {} url={}", task.getClass().getSimpleName(), task.getUrl());

            return result;

        } catch (Exception e) {
            endTime = System.currentTimeMillis();
            long requestCost = endTime - startTime;
            task.setRequestCost(requestCost);
            t = e;
            return onException(task, responseEntity, e);
        } finally {
            boolean retry = task.getRetryNum() != retryCount;
            String retryTag = t != null ? "-" : retry ? task.canRetry() ? "retry" : "limit" : "no";
            metrics.finishedTask(task, t, retryTag);
            onResponse(task, result, t, endTime);
            if (!success) {
                task.getActuallyProxy().fail(System.currentTimeMillis() - startTime);
            }
            SpiderContextHolder.cleanProxy();
        }

    }

    private <T> void onResponse(Task<T> task, T result, Throwable ex, long endTime) {
        if (!(task instanceof AbstractSpiderTask<?>)) {
            return;
        }
        AbstractSpiderTask<?> realTask = ((AbstractSpiderTask<?>) task);
        LimitStrategy limitStrategy = realTask.getLimitStrategy();
        if (limitStrategy == null) {
            return;
        }
        if (!limitStrategy.needLimit(task)) {
            return;
        }
        limitStrategy.afterRequest(task, result, ex, endTime);
    }
}
